不能访问的URL（DEAD LINK）是互联网中非常令人蛋疼的一部分。如果数量很少直接就黏贴进浏览器就完成了，但是如果数量非常多,成千上万条的时候时间和效率就变成宝贵的东西。通过Python来完成这些其实也非常简单，主要使用的lib就是requests，看到200（或者301或者302）就表明网址是正常可以访问；为了方便查找和使用最好最后生成一个表格，并且封装成 命令行程序（使用前记得安装mongodb当做数据库），使用的lib如下：

1. click==6.7
2. requests==2.19.1
3. mongoengine==0.15.0
4. pandas==0.18.1

支持的特性：

1. 不重复请求URL，断线可以直接从没有请求过的url开始
2. 多进程支持
3. 数据库支持
4. 结果输出成excel文件

 

使用方式：

1. 安装Python和相应的lib，并开启mongodb
2. 将想要的检查的url复制到remove.txt，每行一个url
3. 打开get_status.py命令行运行一下命令，运行完毕之后检查urlcheck.csv结果文件

python get_status.py --del no_del 
#or
python get_status.py

#如果输入no_del则不会删除数据库中的数据，可以避免重复请求，选择del则会清空数据库

 

完整代码在：Check_Many_Url_With_Python

#1  读取文件

将需要检查的url放在remove.txt中，每个url一行，为了防止url格式不完整，可以先写个函数转化下url格式

import pandas as pd

def geturllist():
    df = pd.read_table('remove.txt', encoding="ISO-8859-1", names=['url'])
    df.drop_duplicates()
    return df['url'].tolist()


def convert(url):
    if url.startswith('https://www.'):
        return url
    if url.startswith('http://www.'):
        return 'http://' + url[len('http://www.'):]
    if url.startswith('www.'):
        return 'http://' + url[len('www.'):]
    if not url.startswith('http://'):
        return 'http://' + url
    return url

#2  利用requests 发出请求

用requests.get发出请求，返回状态码和最终到达的URL

def get_status_code(url):
    try:
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows; U; Windows NT 6.1; en-US; rv:1.9.1.6) Gecko/20091201 Firefox/3.5.6'
        }
        r = requests.get(convert(url), headers=headers, timeout = 5)
            # print(r.status_code)
        return (str(r.status_code),str(r.url))
        # prints the int of the status code. Find more at httpstatusrappers.com :)
    except requests.ConnectionError:
        #print("failed to connect")
        return "failed to connect"

 

#3 定义数据库模型，将结果存入数据库，并检查请求的URL是否已经请求过

def savestatus(url):
    try:
        if  Domain.objects(url=url).first():
            print('Already in Database')

        else:
            status = get_status_code(convert(url))[0]
            final_url = get_status_code(convert(url))[1]
            Domain(status_code=str(status),
                   url=str(url),
                   final_url=final_url,
                   is_requests=True).save()
            print('{} : {}   has saved \n'.format(str(status), str(url)))


    except:
        print('save fail')


class Domain(DynamicDocument):
    status_code = StringField()
    
    url = StringField()
    final_url = StringField()
    
    is_requests = BooleanField()

 

#4 生成最终的检查excel文件（urlcheck.csv)，并封装成命令行程序

def getcsv():
    d = Domain.objects().all()
    data = [(x.url, x.status_code,x.final_url,x.is_requests) for x in d]
    df1 = pd.DataFrame(data, columns=['url', 'status_code','final_url','is_requests'])
    df1.drop_duplicates()
    df1.to_csv('urlcheck.csv')
    return df1.shape


@click.command()
@click.option('--dele', default='no_del',prompt='( del ) To Delete,( no_del ) To Hold On, Default no_del', help='del Or no_del' )

def deleted(dele):
    if dele == 'del':
        d = Domain.objects().all()
        for x in d:
            d.delete()
        print('delete all records in database')
        return run()
    else:
        return run()

#5 多进程运行程序

def run():
    
    urllist = geturllist()
    pool = mp.Pool(processes=8)
    p = pool.map(savestatus,urllist)
    pool.close()
    pool.join()
    shape = getcsv()



if __name__ == '__main__':
    deleted()